package com.shujia.core

import com.shujia.core.Demo07GroupBy.StuGrp
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo10AggregateByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo10AggregateByKey")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val stuLineRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    // 将每条数据转换成样例类对象
    val stuRDD: RDD[StuGrp] = stuLineRDD.map(line => {
      val splits: Array[String] = line.split(",")
      StuGrp(splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
    })

    /**
     * aggregateByKey：同样需要作用在KV格式的RDD上
     * 需要指定三个参数：
     * zeroValue ： 初始值
     * seqOP：预聚合操作
     * combOp：聚合操作
     */

    // 统计性别人数
    stuRDD
      .map(stu => (stu.gender, 1))
      .aggregateByKey(0L)((l1, i1) => {
        // 预聚合：怎么在Map阶段进行聚合操作
        l1 + i1
      }, (l3, l4) => {
        // 聚合操作：即在Reduce阶段怎么进行聚合
        l3 + l4
      }).foreach(println)

    // 统计班级年龄之和
    stuRDD
      .map(stu => (stu.clazz, stu.age))
      .aggregateByKey(0L)((p1, p2) => {
        p1 + p2
      }, (p3, p4) => {
        p3 + p4
      }).foreach(println)

    // 统计班级年龄最小值
    stuRDD
      .map(stu => (stu.clazz, stu.age))
      .aggregateByKey(100000)((p1, p2) => {
        Math.min(p1, p2)
      }, (p3, p4) => {
        Math.min(p3, p4)
      }).foreach(println)

    // 能不能统计班级平均年龄？
    /**
     * 间接使用预聚合完成AVG的计算
     */
    stuRDD
      .map(stu => (stu.clazz, stu.age))
      .aggregateByKey((0, 0))((t2, v) => {
        val mapSumAge: Int = t2._1 + v
        val mapCnt: Int = t2._2 + 1
        (mapSumAge, mapCnt)
      }, (t2, tt2) => {
        val sumAge: Int = t2._1 + tt2._1
        val totalCnt: Int = t2._2 + tt2._2
        (sumAge, totalCnt)
      }).map {
      case (clazz: String, (sumAge: Int, totalCnt: Int)) =>
        (clazz, sumAge, totalCnt, sumAge / totalCnt.toDouble)
    }.foreach(println)

    // 使用reduceByKey进行简化
    stuRDD
      .map(stu => (stu.clazz, (stu.age, 1)))
      .reduceByKey((t2,tt2)=>{
        // 统计age之和
        val sumAge:Int = t2._1 + tt2._1
        // 统计人数
        val totalCnt:Int = t2._2 + tt2._2
        (sumAge,totalCnt)
      }).map {
      case (clazz: String, (sumAge: Int, totalCnt: Int)) =>
        (clazz, sumAge, totalCnt, sumAge / totalCnt.toDouble)
    }.foreach(println)

  }

}
